package PVandUv;

import bean.UserBehavior;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import java.util.HashMap;


/**
 * @author Spring_Hu
 * @date 2021/10/12 21:10
 */
public class UV {
    public static void main(String[] args) throws Exception {

        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        DataStreamSource<String> ds = env.readTextFile("test/UserBehavior.csv");
        env.setParallelism(2);
        ds.map(value -> new UserBehavior(Long.parseLong(value.split(",")[0]),
                Long.parseLong(value.split(",")[1]),Integer.parseInt(value.split(",")[2]),
                value.split(",")[3],Long.parseLong(value.split(",")[4]))).returns(Types.GENERIC(UserBehavior.class))
                .filter(value -> "pv".equals(value.getBehavior())).map(value -> (Tuple2.of(value.getUserId().toString(),1L)))
                .returns(Types.TUPLE(Types.STRING,Types.LONG))
                .map(new MapFunction<Tuple2<String, Long>, Long>() {
                    HashMap<String, Long> result = new HashMap<String, Long>();
                    @Override
                    public Long map(Tuple2<String, Long> value) throws Exception {
                        result.put(value.f0,value.f1);
                        return Long.valueOf(result.size());
                    }
                }).setParallelism(1)
                .print();
      env.execute();

    }
}
